package com.monkeyboy.demo.base.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 消息消费者
 *
 * @Author Gavin
 * @date 2020.08.18 12:54
 */
public class Consumer1 {
    public static void main(String[] args) throws Exception {
//        1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
//        2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.1.140:9876;192.168.1.141:9876");
//        3.订阅主题Topic和Tag(第二个参数就是tag,支持*)
        consumer.subscribe("test", "*");
        //设定消费模式：负载均衡(默认)|广播模式
//        consumer.setMessageModel(MessageModel.CLUSTERING);
        //设定消费模式：负载均衡(默认)|广播模式
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("【"+Thread.currentThread().getName()+"】  "+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功或者失败后要给一个反馈
            }
        });
//        5.启动消费者consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
